package com.jili.reactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.time.Duration;

/**
 *  ClassName: FluDemo
 *  Package com.jili.reactor
 *  Description:
 *  @Author: 廖继礼
 *  @Create: 2025/3/22
 *  @Version: v1.0
 *
 */
public class FluDemo {

    public static void main(String[] args) throws InterruptedException {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6, 7)
                .delayElements(Duration.ofSeconds(1))
                .doOnComplete(() -> {
                    System.out.println("流正常结束");
                }).doOnCancel(() -> {
                    System.out.println("取消流");
                }).doOnError(throwable -> {
                    System.out.println("流出错" + throwable);
                }).doOnNext(item -> {
                    System.out.println("item arrive: " + item);
                });
        flux.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("发布这和订阅者绑定");
                request(1);
            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("元素到达：" + value);
                if (value < 6) {
                    if (value == 3) {
                        int result = value / 0;
                    }
                    request(1);
                } else {
                    cancel();
                }
            }

            @Override
            protected void hookOnComplete() {
                super.hookOnComplete();
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                super.hookOnError(throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("结束了");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("结束信号");
            }
        });
        Thread.sleep(20000);
    }
}
